package cn.funeralobjects.youtube.dl.extension.service.impl;

import cn.funeralobjects.youtube.dl.extension.enums.TaskStatusType;
import cn.funeralobjects.youtube.dl.extension.exec.YoutubeDl;
import cn.funeralobjects.youtube.dl.extension.model.DownloadTask;
import cn.funeralobjects.youtube.dl.extension.model.DownloadingStatus;
import cn.funeralobjects.youtube.dl.extension.model.Running;
import cn.funeralobjects.youtube.dl.extension.model.TaskStatus;
import cn.funeralobjects.youtube.dl.extension.repository.TaskRepository;
import cn.funeralobjects.youtube.dl.extension.service.DownloadService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.File;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @author FuneralObjects
 * Create date: 2020/4/3 2:55 PM
 */
@Service
public class DownloadServiceImpl implements DownloadService {

    @Resource
    private TaskRepository completedTaskRepository;
    @Resource
    private TaskRepository downloadingTaskRepository;
    @Resource
    private TaskRepository interruptTaskRepository;

    private static final Logger LOGGER = LoggerFactory.getLogger(DownloadService.class);
    /**
     * 最大线程数
     */
    private static final Integer MAX_THREADS = 2;
    /**
     * 任务回调缓存
     */
    private static final ConcurrentMap<String, Consumer<TaskStatus>> TASK_CONSUMER = new ConcurrentHashMap<>();
    /**
     * 执行的线程集合
     */
    private static final ConcurrentMap<String, Running> RUNNING_MAP = new ConcurrentHashMap<>();
    /**
     * 任务集合
     */
    private static final ConcurrentMap<String, DownloadTask> TASK_MAP = new ConcurrentHashMap<>();
    /**
     * 执行中的集合
     */
    private static final Set<String> RUNNING_SET = ConcurrentHashMap.newKeySet();
    /**
     * 等待中执行集合
     */
    private static final LinkedList<String> WAITING_LIST = new LinkedList<>();
    private static final AtomicInteger RUNNING_COUNT_DOWN = new AtomicInteger(0);
    private static final ThreadFactory THREAD_FACTORY;
    private static final ExecutorService POOL_EXECUTOR;
    private static Integer index = 0;

    static {
        THREAD_FACTORY = r -> {
            index++;
            Thread result = new Thread(r);
            result.setName("download-thread-" + index);
            return result;
        };
        POOL_EXECUTOR = new ThreadPoolExecutor(MAX_THREADS + 1, MAX_THREADS + 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy());
    }

    private synchronized void startDownloading() {
        if (MAX_THREADS.compareTo(RUNNING_COUNT_DOWN.get()) <= 0) {
            return;
        }
        if (WAITING_LIST.size() <= 0) {
            return;
        }
        startNext();
    }

    private synchronized void complete(String key){
        DownloadTask task = stopAndGet(key);
        if(task == null){
            return;
        }
        completedTaskRepository.push(task);
        downloadingTaskRepository.remove(task.getId(), task.getFormat());
    }

    private synchronized void interrupt(String key){
        DownloadTask task = stopAndGet(key);
        if(task == null){
            return;
        }
        interruptTaskRepository.push(task);
        downloadingTaskRepository.remove(task.getId(), task.getFormat());
    }

    private DownloadTask stopAndGet(String key){
        DownloadTask task = TASK_MAP.get(key);
        if(task == null){
            return null;
        }
        task.setEndTime(System.currentTimeMillis());
        TASK_MAP.remove(key);
        RUNNING_SET.remove(key);
        Running running = RUNNING_MAP.get(key);
        if(running != null){
            running.stop();
            RUNNING_MAP.remove(key);
        }
        return task;
    }

    private synchronized void running(String key){
        DownloadTask task = TASK_MAP.get(key);
        Running running = StringUtils.isEmpty(task.getProxy())
                ? YoutubeDl.downloadInSpecialFolder(task.getId(), task.getFormat(), DOWNLOAD_FOLDER_PATH)
                : YoutubeDl.downloadInSpecialFolder(task.getId(), task.getFormat(), task.getProxy(), DOWNLOAD_FOLDER_PATH);
        running.setNow("Waiting...");
        RUNNING_MAP.put(key, running);
        running.setCallback(result -> {
            if (result) {
                complete(key);
            } else {
                interrupt(key);
            }
            RUNNING_COUNT_DOWN.decrementAndGet();
            consumerRun(result ? TaskStatusType.COMPLETED : TaskStatusType.INTERRUPTED, task);
            startDownloading();
        });
        RUNNING_SET.add(key);
        RUNNING_COUNT_DOWN.incrementAndGet();
        running.execute(POOL_EXECUTOR);
        consumerRun(TaskStatusType.RUNNING, task);
    }

    private synchronized void startNext() {
        String first = WAITING_LIST.pop();
        running(first);
    }

    private static String getKey(String id, String format) {
        return id + " (" + format + ")";
    }

    private static void checkFile() {
        File downloadFolder = new File(DOWNLOAD_FOLDER_PATH);
        if(!downloadFolder.exists() || downloadFolder.isFile()){
            if(!downloadFolder.mkdirs()){
                throw new RuntimeException("The folder["+DOWNLOAD_FOLDER_PATH+"] creation failed!");
            }
        }
    }

    @Override
    public List<DownloadTask> getCompleted() {
        return completedTaskRepository.getAllTasks();
    }

    @Override
    public DownloadingStatus getDownloadingStatus() {
        return new DownloadingStatus()
                .setRunning(RUNNING_SET.parallelStream().map(TASK_MAP::get).collect(Collectors.toList()))
                .setInterrupt(interruptTaskRepository.getAllTasks())
                .setWaiting(WAITING_LIST.parallelStream().map(TASK_MAP::get).collect(Collectors.toList()));
    }

    @Override
    public String getCurrentLog(String id, String format) {
        String key= getKey(id, format);
        if(RUNNING_MAP.containsKey(key)){
            return RUNNING_MAP.get(key).getNow();
        }
        return null;
    }

    @Override
    public void addDownloadTask(String id, String format, String title, String proxy) {
        String key = getKey(id, format);
        if (TASK_MAP.containsKey(key) || getCompleted().parallelStream().anyMatch(task -> getKey(task.getId(), task.getFormat()).equals(key))) {
            return;
        }
        if (completedTaskRepository.getAllTasks().parallelStream().anyMatch(task -> getKey(task.getId(), task.getFormat()).equals(key))) {
            return;
        }
        DownloadTask task = new DownloadTask()
                .setFormat(format)
                .setTitle(title)
                .setId(id)
                .setProxy(proxy)
                .setStartTime(System.currentTimeMillis());
        downloadingTaskRepository.push(task);
        this.addTask(task);
    }

    @Override
    public void addNextConsumer(String key, Consumer<TaskStatus> consumer) {
        if(TASK_CONSUMER.containsKey(key)){
            return;
        }
        TASK_CONSUMER.put(key, consumer);
    }

    @Override
    public void removeConsumer(String key) {
        TASK_CONSUMER.remove(key);
    }

    @Override
    public void retryInterrupt(String id, String format) {
        DownloadTask task = interruptTaskRepository.getByIdAndFormat(id, format);
        if (task == null) {
            return;
        }
        interruptTaskRepository.remove(id, format);
        downloadingTaskRepository.push(task);
        this.addTask(task);
    }

    @Override
    public void stop(String id, String format) {
        String key = getKey(id, format);
        if (!TASK_MAP.containsKey(key)) {
            return;
        }
        if (!RUNNING_MAP.containsKey(key)) {
            consumerRun(TaskStatusType.INTERRUPTED, TASK_MAP.get(key));
            interrupt(key);
            return;
        }
        RUNNING_MAP.get(key).setStopped(true);
    }

    @Override
    public void removeInterrupt(String id, String format) {
        interruptTaskRepository.remove(id, format);
    }


    private synchronized static void consumerRun(TaskStatusType status, DownloadTask task) {
        LOGGER.info("Task Status {}: {} - {}", status.name(), task.getId(), task.getFormat());
        TASK_CONSUMER.values().parallelStream().forEach(consumer -> consumer.accept(
                new TaskStatus()
                        .setType(status)
                        .setTask(task)
                        .setRunningCount(RUNNING_COUNT_DOWN.get())
        ));
    }

    @PostConstruct
    private void init() {
        checkFile();
        this.readDownloading();
    }

    private void readDownloading() {
        downloadingTaskRepository.getAllTasks().forEach(this::addTask);
    }


    private void addTask(DownloadTask task){
        String key = getKey(task.getId(),task.getFormat() );
        TASK_MAP.put(key, task);
        WAITING_LIST.addLast(key);
        consumerRun(TaskStatusType.WAITING, task);
        startDownloading();
    }

}
